#include "EventLoop.h"
#include "Channel.h"
#include "Poller.h"
#include "Logger.h"

#include <sys/eventfd.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <memory>

// 避免一个线程创建多个EventLoop
__thread EventLoop *t_loopInThisThread = nullptr; // thread_local 线程独享变量

// Poller IO复用接口超时时间
const int kPollTimeMs = 10000;

/**
 * @brief 创建wakeupfd，用来notify唤醒subReactor处理新来的channel
 *
 * @return int
 */
int createEventfd()
{
    // eventfd()创建一个事件对象, 用来实现，进程(线程)间的等待/通知(wait/notify) 机制
    // EFD_NONBLOCK设置对象为非阻塞状态,EFD_CLOEXE调用exec后会自动关闭文件描述符，防止泄漏
    int evtfd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0)
    {
        LOG_FATAL("eventfd error:%d\n", errno);
    }
    return evtfd;
}

EventLoop::EventLoop()
    : looping_(false), quit_(false), callingPendingFunctors_(false), threadId_(CurrentThread::tid()), poller_(Poller::newDefaultPoller(this)), wakeupFd_(createEventfd()), wakeupChannel_(new Channel(this, wakeupFd_))
{
    LOG_DEBUG("EventLoop create %p in thread %d\n", this, threadId_);
    // 判断线程是否重复创建EventLoop，保证one eventloop per thread
    if (t_loopInThisThread)
    {
        LOG_FATAL("Another EventLoop %p exists in this thread %d\n", t_loopInThisThread, threadId_);
    }
    else
    {
        t_loopInThisThread = this;
    }
    // 封装并设置wakeFd_的事件回调，用户线程间的notify
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
    // 监听读事件
    wakeupChannel_->enableReading();
}

EventLoop::~EventLoop()
{
    wakeupChannel_->disableAll(); // 关闭线程间notify文件
    wakeupChannel_->remove();
    close(wakeupFd_);
    t_loopInThisThread = nullptr; // 重置
}
// 开启/退出事件循环
void EventLoop::loop()
{
    looping_ = true;
    quit_ = false;

    LOG_INFO("EventLoop %p start looping\n", this);

    while (!quit_)
    {
        activeChannels_.clear();
        // 获取就绪Channel队列
        pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
        for (Channel *channel : activeChannels_)
        { // 执行event事件回调
            channel->handlEvent(pollReturnTime_);
        }
        // 执行传递的回调函数
        doPendingFunctors();
    }

    LOG_INFO("EventLoop %p stop looping\n", this);
    looping_ = false;
}
void EventLoop::quit()
{
    quit_ = true;
    if (!isInLoopThread()) // 当前loop不在所属的线程
    {
        wakeup(); // notify线程
    }
}

// 在当前loop中执行回调函数
void EventLoop::runInLoop(Functor cb)
{
    if (isInLoopThread()) // 在所属线程中直接执行
    {
        cb();
    }
    else
    {
        queueInLoop(cb); // 不在所属线程放入queue
    }
}
// 将回调函数放入队列，唤醒loop的线程进行执行
void EventLoop::queueInLoop(Functor cb)
{
    {
        std::unique_lock<std::mutex> lock(mutex_);
        pendingFunctors_.emplace_back(cb); // 储存
    }

    // 所属线程阻塞或者正在执行回调，都进行notify
    if (!isInLoopThread() || callingPendingFunctors_)
    {
        wakeup();
    }
}
// 处理notify
void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR("EventLoop::wakeup() reads %lu bytes instead of 8\n", n);
    }
}

// notify loop所在线程-唤醒
void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = write(wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8\n", n);
    }
}
/**
 * @brief 调用IO复用更新事件状态
 *
 * @param channel
 */
void EventLoop::updateChannel(Channel *channel)
{
    poller_->updateChannel(channel);
}
void EventLoop::removeChannel(Channel *channel)
{
    poller_->removeChannel(channel);
}
bool EventLoop::hasChannel(Channel *channel)
{
    return poller_->hasChannel(channel);
}
/**
 * @brief 执行回调函数
 *
 */
void EventLoop::doPendingFunctors()
{
    std::vector<Functor> functors;
    callingPendingFunctors_ = true; // 正在执行回调函数

    // 取出回调函数队列数据，避免添加回调函数响应过久
    {
        std::unique_lock<std::mutex> lock(mutex_);
        functors.swap(pendingFunctors_);
    }

    for (const Functor &functor : functors)
    {
        functor();
    }
    callingPendingFunctors_ = false;
}